package cn.iotab.iotkit.transport.mqtt;

import cn.iotab.iotkit.transport.mqtt.handler.ConnectHandler;
import cn.iotab.iotkit.transport.mqtt.handler.DisconnectHandler;
import cn.iotab.iotkit.transport.mqtt.handler.PingHandler;
import cn.iotab.iotkit.transport.mqtt.handler.PublishHandler;
import cn.iotab.iotkit.transport.mqtt.handler.SubscribeHandler;
import cn.iotab.iotkit.transport.mqtt.handler.UnsubscribeHandler;
import cn.iotab.iotkit.transport.mqtt.session.MqttSessionCtx;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.UUID;

/**
 * @author Andrew Shvayka
 */
@Slf4j
public class MqttTransportHandler extends ChannelInboundHandlerAdapter implements GenericFutureListener<Future<? super Void>> {

    private final String sessionId;

    private final MqttSessionCtx deviceSessionCtx;

    private final ConnectHandler connectHandler;
    private final PublishHandler publishHandler;
    private final SubscribeHandler subscribeHandler;
    private final UnsubscribeHandler unsubscribeHandler;
    private final PingHandler pingHandler;
    private final DisconnectHandler disconnectHandler;

    MqttTransportHandler(MqttTransportContext context, SslHandler sslHandler) {
        this.sessionId = UUID.randomUUID().toString();

        this.deviceSessionCtx = new MqttSessionCtx(sessionId, context);

        //
        this.connectHandler = new ConnectHandler(context, deviceSessionCtx, sslHandler);
        this.publishHandler = new PublishHandler(context, deviceSessionCtx);
        this.subscribeHandler = new SubscribeHandler(context, deviceSessionCtx);
        this.unsubscribeHandler = new UnsubscribeHandler(context, deviceSessionCtx);
        this.pingHandler = new PingHandler(context, deviceSessionCtx);
        this.disconnectHandler = new DisconnectHandler(context, deviceSessionCtx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        log.trace("[{}] Processing msg: {}", sessionId, msg);
        try {
            if (msg instanceof MqttMessage) {
                MqttMessage message = (MqttMessage) msg;
                if (message.decoderResult().isSuccess()) {
                    processMqttMsg(ctx, message);
                } else {
                    log.error("[{}] Message processing failed: {}", sessionId, message.decoderResult().cause().getMessage());
                    ctx.close();
                }
            } else {
                ctx.close();
            }
        } finally {
            ReferenceCountUtil.safeRelease(msg);
        }
    }

    /**
     * 处理收到的MQTT消息
     *
     * @param ctx
     * @param msg
     */
    private void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
        deviceSessionCtx.setChannelHandlerContext(ctx);
        //无效消息
        if (msg.fixedHeader() == null) {
            InetSocketAddress address = (InetSocketAddress) ctx.channel().remoteAddress();
            log.info("[{}:{}] Invalid message received", address.getHostName(), address.getPort());
            deviceSessionCtx.processDisconnect();
            return;
        }
        processRegularSessionMsg(ctx, msg);

    }


    private void processRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {
        MqttMessageType messageType = msg.fixedHeader().messageType();
        switch (messageType) {
            case CONNECT:
                connectHandler.process(ctx, (MqttConnectMessage) msg);
                break;
            case PUBLISH:
                publishHandler.process(ctx, (MqttPublishMessage) msg);
                break;
            case PUBACK:
                //ignore
                break;
            case SUBSCRIBE:
                subscribeHandler.process(ctx, (MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:
                unsubscribeHandler.process(ctx, (MqttUnsubscribeMessage) msg);
                break;
            case PINGREQ:
                pingHandler.process(ctx, msg);
                break;
            case DISCONNECT:
                disconnectHandler.process(ctx, msg);
                break;
            default:
                log.warn("ignore unsupported mqttMessage type:{}", messageType.name());
                break;
        }
    }


    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("[{}] Unexpected Exception", sessionId, cause);
        ctx.close();
    }


    @Override
    public void operationComplete(Future<? super Void> future) throws Exception {
        log.info("operationComplete, will close channel");
        deviceSessionCtx.processDisconnect();
    }
}
